
package org.shj.spark.operator;

import java.util.Arrays;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;

/**
 * reduceByKey = groupByKey + reduce
 * shuffle 洗牌 = map 端 + reduce 端
 * spark 里的reduceByKey在map端自带combiner
 * 
 * @author Administrator
 *
 */
public class ReduceByKeyOperator {

	public static void main(String[] args) {
		SparkConf conf = new SparkConf().setMaster("local").setAppName("ReduceByKeyOperator");
		JavaSparkContext sc = new JavaSparkContext(conf);
		
		List<Tuple2<String, Integer>> scoreList = Arrays.asList(new Tuple2<String, Integer>("xuruyun", 100), 
				new Tuple2<String, Integer>("liangyongqi", 150),
				new Tuple2<String, Integer>("zhuyin", 160),
				new Tuple2<String, Integer>("zhuyin", 180));
		
		JavaPairRDD<String, Integer> rdd = sc.parallelizePairs(scoreList);
		
		rdd.reduceByKey(new Function2<Integer, Integer, Integer>(){
			private static final long serialVersionUID = 6460708222395484939L;

			public Integer call(Integer v1, Integer v2) throws Exception {
				return v1 + v2;
			}
			
		}).foreach(new VoidFunction<Tuple2<String, Integer>>(){
			private static final long serialVersionUID = -4179994634791329847L;

			public void call(Tuple2<String, Integer> t) throws Exception {
				System.out.println(t._1 + " " + t._2);
			}
			
		});
		
		sc.close();
		
	}
}
